package com.apex.spark.sink;

import com.apex.spark.SparkEnvironment;
import com.apex.spark.batch.SparkBatchSink;
import com.apex.spark.stream.SparkStreamingSink;
import com.apex.spark.structuredstream.SparkStructuredStreamingSink;
import com.typesafe.config.Config;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;

public class ConsoleSink implements SparkBatchSink, SparkStreamingSink, SparkStructuredStreamingSink {
    private Config config;

    @Override
    public void outputBatch(SparkEnvironment env, Dataset<Row> dataSet) {
        dataSet.show();
    }

    @Override
    public void outputStreaming(SparkEnvironment env, Dataset<Row> dataSet) {
    }

    @Override
    public void outputStructuredStreaming(SparkEnvironment env, Dataset<Row> dataSet) {
        StreamingQuery query = dataSet.writeStream()
                .outputMode("complete")
                .format("console")
                .start();

        try {
            query.awaitTermination();
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Override
    public void prepare(SparkEnvironment plugin) {

    }

    @Override
    public Config getConfig() {
        return config;
    }

    @Override
    public void setConfig(Config config) {
        this.config = config;
    }

}
